package com.atguigu.flink.sql.join;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by Smexy on 2023/4/14
 *
 *
 *      维表关联。
 *          把事实表做成动态表。 让事实表中没到达的每一条数据都去 关联维度表中的数据。
 *
 *          维度表： 读某个数据库，进行访问。
 *
 *          本质： 单表关联。
 */
public class Demo3_LoopUpJoin
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        env.setParallelism(1);


        SingleOutputStreamOperator<WaterSensor> ds1 = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction());

        Schema schema = Schema.newBuilder()
                              .column("id", "STRING")
                              .column("ts", "BIGINT")
                              .column("vc", "INT")
                              //事实表中，必须有 处理时间
                              .columnByExpression("pt", "proctime()")
                              .columnByExpression("et", "TO_TIMESTAMP_LTZ(ts,3)")
                              .watermark("et","et - INTERVAL '0.001' SECOND")
                              .build();
        //准备事实表
        tableEnvironment.createTemporaryView("t1",ds1,schema);

        //映射已经存在的维度表
        String createTableSql = " create table t2 ( id STRING, ts BIGINT , vc INT " +
            "    )WITH (" +
            "  'connector' = 'jdbc'," +
            "   'url' = 'jdbc:mysql://hadoop102:3306/221109'," +
            "   'table-name' = 'ws' ," +
            "   'username' = 'root' ," +
            "   'password' = '000000' " +
            ") ";

        tableEnvironment.executeSql(createTableSql);



        tableEnvironment
            .sqlQuery("  select t1.id ,t1.vc ,t2.id ,t2.vc " +
            "   from  t1 join t2 " +
            // 固定写法: FOR SYSTEM_TIME AS OF o.proc_time
            "   FOR SYSTEM_TIME AS OF t1.pt " +
            "  on t1.id = t2.id ")
            .execute()
            .print();




    }
}
